package model;

import java.io.IOException;
import java.io.StringReader;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.lucene.analysis.TokenStream;
import org.apache.lucene.analysis.core.StopFilter;
import org.apache.lucene.analysis.fr.FrenchAnalyzer;
import org.apache.lucene.analysis.standard.ClassicTokenizer;
import org.apache.lucene.analysis.tokenattributes.CharTermAttribute;
import org.apache.lucene.util.Version;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;

public class TwitterMapper extends Mapper<LongWritable,Text,Text,Text> {
	
	private FrenchAnalyzer analyzer;

	  @Override
	  protected void map(LongWritable key, Text value, Context context) {
	    String text;
	    Document doc = Jsoup.parse(value.toString());
    	doc.body().select("script, jscript").remove();
	    text = tokenize(value);
	    Text nullWrite = new Text(NullWritable.get().toString());
	    try {
			context.write(nullWrite, new Text(text));
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	    
	  }
	  
	  @Override
	  protected void setup(Context context) throws IOException, InterruptedException {
		  this.analyzer = new FrenchAnalyzer(Version.LUCENE_43);
	  }
	  
	  
	  public String tokenize (Text value) 
	  {
		  TokenStream tokStr = new ClassicTokenizer(Version.LUCENE_43,new StringReader(value.toString()));
		  tokStr = new StopFilter(Version.LUCENE_43, tokStr, FrenchAnalyzer.getDefaultStopSet());
		  
		  StringBuilder builder = new StringBuilder();
		  CharTermAttribute token = tokStr.getAttribute(CharTermAttribute.class);
		  
		  try 
		  {
			  while(tokStr.incrementToken())
			  {
				  if(builder.length() > 0)
					  builder.append(" ");
				  
				  builder.append(token.toString());
			  }
		  } 
		  catch(IOException e)
		  {
			  e.printStackTrace();
		  }
		  
		  return builder.toString();
	  }
}
